package cn.ac.iie.bdb_to_mongo;

import cn.ac.iie.bdb_to_mongo.bdbhandler.BDBUtil;
import cn.ac.iie.bdb_to_mongo.bdbhandler.WriteBdbTask;
import cn.ac.iie.bdb_to_mongo.commons.Constants;
import cn.ac.iie.bdb_to_mongo.commons.KillHandler;
import cn.ac.iie.bdb_to_mongo.mongohandler.UpdateMongoTask;
import cn.ac.iie.bdb_to_mongo.receiver.MessageFailReceiver;
import cn.ac.iie.bdb_to_mongo.receiver.MessageSuccessReceiver;
import cn.ac.iie.bdb_to_mongo.sender.MessageSenderThread;
import cn.ac.iie.bdb_to_mongo.sender.RewriteSenderThread;
import cn.ac.iie.di.datadock.rdata.exchange.client.exception.REConnectionException;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.ConsumePosition;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.REMessage;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.connection.REConnection;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.session.REReceiveSession;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.session.REReceiveSessionBuilder;
import org.apache.commons.configuration2.FileBasedConfiguration;
import org.apache.commons.configuration2.PropertiesConfiguration;
import org.apache.commons.configuration2.builder.FileBasedConfigurationBuilder;
import org.apache.commons.configuration2.builder.fluent.Parameters;
import org.apache.commons.configuration2.convert.DefaultListDelimiterHandler;
import org.apache.commons.configuration2.ex.ConfigurationException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Timer;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.atomic.AtomicBoolean;


/**
 * @auther Fighter Created on 2016/12/1.
 */

public class BdbMain {

    private static final Logger LOG = LoggerFactory.getLogger(BdbMain.class);
    public static LinkedBlockingDeque<REMessage> newValQueue;//写mongo

    public static LinkedBlockingDeque<REMessage> toMqQueue1;//回写旧表MQ
    public static LinkedBlockingDeque<REMessage> toMqQueue2;//回写旧表MQ

    public static LinkedBlockingDeque<REMessage> reWriteQueue;//mongo不存在则回写MQ


    private static final String PROPERTIES_PATH;
    public static BDBUtil<String, Long> testDB;

    //读取配置文件
    public static FileBasedConfiguration configuration;
    //标志量
    public static AtomicBoolean writeFlag = new AtomicBoolean(true);//写bdb标志量
    public static AtomicBoolean updateMongoFlag = new AtomicBoolean(true);//更新mongo标志量

    static {

        PROPERTIES_PATH = ClassLoader.getSystemClassLoader().getResource("configuration.properties").getFile();
        if (PROPERTIES_PATH == null || PROPERTIES_PATH.isEmpty()) {
            throw new RuntimeException("properties path is empty!");
        } else {
            LOG.info("PROPERTIES_PATH = {}", PROPERTIES_PATH);
        }

        try {
            configuration = initConfiguration();
        } catch (ConfigurationException e) {
            LOG.error("initConfiguration error!");
        }
    }

    public static void main(String[] args) {

        try {
            testDB = BDBUtil.getInstance(configuration);//初始化bdb数据库
            //启动发送队列线程,旧表队列
            toMqQueue1 = new LinkedBlockingDeque<>(50);
            Thread sendToMq1 = new Thread(new MessageSenderThread(
                    configuration.getString(Constants.RECEIVER_MQ_NAMESERVER),
                    configuration.getString(Constants.SENDER_MQ_SEND_TOPIC1),
                    toMqQueue1),
                    "sendToMq thread1");
            sendToMq1.start();

            //启动发送队列线程,旧表队列
            toMqQueue2 = new LinkedBlockingDeque<>(50);
            Thread sendToMq2 = new Thread(new MessageSenderThread(
                    configuration.getString(Constants.RECEIVER_MQ_NAMESERVER),
                    configuration.getString(Constants.SENDER_MQ_SEND_TOPIC2),
                    toMqQueue2
            ),
                    "sendToMq thread2");
            sendToMq2.start();

            //启动写BDB线程
            newValQueue = new LinkedBlockingDeque<>(50);
            int writeBdbThreadNum = configuration.getInt(Constants.WRITE_BDB_THREAD_NUM, 10);
            for (int i = 0; i < writeBdbThreadNum; i++) {
                Thread t = new Thread(new WriteBdbTask());
                t.start();
            }

            reWriteQueue = new LinkedBlockingDeque<>(50);
            Thread reWriteThread = new Thread(
                    new RewriteSenderThread(
                            configuration.getString(Constants.RECEIVER_MQ_NAMESERVER),
                            configuration.getString(Constants.RECEIVER_MQ_READ_TOPIC),
                            reWriteQueue
                            )
            );
            reWriteThread.start();

            //启动接收数据线程
            String mqNameserver = configuration.getString(Constants.RECEIVER_MQ_NAMESERVER);
            String mqTopic = configuration.getString(Constants.RECEIVER_MQ_READ_TOPIC);
            String mqGroup = configuration.getString(Constants.RECEIVER_MQ_GROUP);
            int consumerThreadNum = configuration.getInt(Constants.RECEIVER_THREAD_NUM, 4);

            LOG.info("--------------------------------------------");
            LOG.info("   receiver mq nameserver = " + mqNameserver);
            LOG.info("        receiver mq topic = " + mqTopic);
            LOG.info("       receiver  mq group = " + mqGroup);
            LOG.info("      receiver thread num = " + consumerThreadNum);
            LOG.info("--------------------------------------------");

            REConnection conn = new REConnection(mqNameserver);
            REReceiveSessionBuilder builder = (REReceiveSessionBuilder) conn.getReceiveSessionBuilder(mqTopic);

            /*//测试!!!!!!!!!!
            boolean test = true;

            if (test) {
                LOG.info("consumer from last offset!");
                builder.setConsumPosition(ConsumePosition.CONSUME_FROM_FIRST_OFFSET);
            } else {
                //测试！！！！！！！！！！！
                builder.setConsumPosition(ConsumePosition.CONSUME_FROM_TIMESTAMP);
                SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
                //测试!!!!!!!!!!!!!!!!!!!
                Date date = sdf.parse("2016-12-05");
                builder.setConsumeTimestamp(date.getTime());
            }*/

            if (args.length == 0) {
                LOG.info("consumer from last offset!");
                builder.setConsumPosition(ConsumePosition.CONSUME_FROM_LAST_OFFSET);
            } else {
                builder.setConsumPosition(ConsumePosition.CONSUME_FROM_TIMESTAMP);
                SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
                Date date = sdf.parse(args[0]);
                LOG.info("consumer from date {} ", args[0]);
                builder.setConsumeTimestamp(date.getTime());
            }

            builder.setConsumeThreadNum(consumerThreadNum);
            builder.setHandler(new MessageSuccessReceiver(newValQueue));
            builder.setFailureHandler(new MessageFailReceiver());
            builder.setGroupName(mqGroup);
            REReceiveSession session = (REReceiveSession) builder.build();
            session.start();

            //启动更新Mongo定时任务:每20分钟更新一次mongo
            Timer timer = new Timer();
            timer.schedule(new UpdateMongoTask(), configuration.getLong(Constants.MONGO_UPDATE_DELAY), configuration.getLong(Constants.MONGO_UPDATE_INTERVAL));
//            timer.schedule(new UpdateMongoTask(),30000,30000);

            //启动更新Mongo定时任务:每20分钟更新一次mongo
            Timer timer2 = new Timer();
            timer2.schedule(new UpdateMongoTask(), configuration.getLong(Constants.MONGO_UPDATE_DELAY), configuration.getLong(Constants.MONGO_UPDATE_INTERVAL));

            //程序关闭
            KillHandler killHandler = new KillHandler();
            killHandler.registerSignal("TERM");
            killHandler.putQueue("newValQueue", newValQueue);
            killHandler.putQueue("toMqQueue1", toMqQueue1);
            killHandler.putQueue("toMqQueue2", toMqQueue2);
            killHandler.setTimer(timer);
            killHandler.setConn(conn);

        } catch (ParseException | REConnectionException e) {
            LOG.error("start program error!", e);
            System.exit(1);
        }
    }

    private static FileBasedConfiguration initConfiguration() throws ConfigurationException {
        Parameters params = new Parameters();
        FileBasedConfigurationBuilder<FileBasedConfiguration> confBuilder
                = new FileBasedConfigurationBuilder<FileBasedConfiguration>(
                PropertiesConfiguration.class)
                .configure(params.properties()
                        .setFileName(PROPERTIES_PATH)
                        .setListDelimiterHandler(new DefaultListDelimiterHandler(','))
                        .setThrowExceptionOnMissing(true));
        FileBasedConfiguration configuration = confBuilder.getConfiguration();
        return configuration;
    }
}